//++
//
// Copyright (c) Microsoft Corporation.  All rights reserved.
//
// Module Name:
//
//  DispatcherService.cs
//    
// Abstract:
//
//  The Dispatcher service class. This is the main entry point for the service.
//  
//--

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Messaging;
using System.ServiceProcess;
using System.Threading;
using Microsoft.OfficeCommunicationsServer.Applications.Common;
using Microsoft.Rtc.Collaboration;

namespace Microsoft.EnhancedReminders.Dispatcher
{
    /// <summary>
    /// Windows service class for delivering enhanced reminders
    /// </summary>
    public class DispatcherService : ServiceBase
    {
        private MessageQueue _queue;
        private DispatcherClassFactory _factory;
        private AutoResetEvent _waitEvent;
        private NotificationDispatcher _dispatcher;
        private object _stateSyncRoot = new object();
        private int _refCount;
        private bool _stopping;

        private bool _connected; // whether the service is connected to OCS and notifications should be handled.
        private bool _outstandingRead;

        // The dispatcher service maps each notification ID to the response queue path to write
        // a NotificationResult to. If in the future we find there is additional data we think
        // we need to store for each NotificationItem, consider refactoring the code so that
        // the Dispatcher namespace has its own class that wraps NotificationItem and carries this
        // extra data instead.
        // This may be a possible solution to the filename issue in DispatcherClassFactory as well.
        private IDictionary<NotificationId, string> _responseLocations;

        /// <summary>
        /// Starts the Dispatcher service by creating the SIP endpoint used by
        /// various classes and begins listening to MSMQ for messages to deliver.
        /// </summary>
        protected override void OnStart(string[] args)
        {
            // Initialize logging for the Dispatcher
            EnhancedRemindersHelpers.InitializeLogging("Microsoft Enhanced Reminders Dispatcher");

            OperationalLogs.LogServiceStartingInformation();

            try
            {
                _factory = new DispatcherClassFactory();
                _dispatcher = new NotificationDispatcher(_factory);
                _responseLocations = new Dictionary<NotificationId, string>();
                _queue = new MessageQueue(Helpers.GetSetting<string>("NotificationQueueServer",null));

                _factory.RegistrationStateChanged += RegistrationStateChanged; 
                _factory.Initialize();
                _dispatcher.DeliverNotificationCompleted += OnNotificationCompleted;
            }
            catch (Exception ex)
            {
                OperationalLogs.LogServiceStartError(ex);
                throw;
            }

            OperationalLogs.LogServiceStartedInformation();
        }

        /// <summary>
        /// Tracks the registration state of the endpoint used for delivering messages. If the endpoint loses its
        /// connection, we should try to reconnect.
        /// </summary>
        private void RegistrationStateChanged(object sender, LocalEndpointStateChangedEventArgs e)
        {
            if ((e.State == LocalEndpointState.Terminated || e.State == LocalEndpointState.Reestablishing) && !_stopping)
            {
                OperationalLogs.LogOcsConnectionWarning();
                _connected = false;
            }
            else if (e.State == LocalEndpointState.Established)
            {
                // Sometimes the state changed event is fired even though we're already established. To keep us from
                // spamming the event log, only execute the following if we're not connected.
                if (!_connected)
                {
                    OperationalLogs.LogReconnectedToOcs();
                    _connected = true;

                    lock (_stateSyncRoot)
                    {
                        if (!_outstandingRead)
                        {
                            _outstandingRead = true;
                        }
                        else
                        {
                            return;
                        }
                    }

                    try
                    {
                        _queue.BeginReceive(TimeSpan.FromMinutes(1.0), this, new AsyncCallback(OnMessageReceived));
                    }
                    catch (MessageQueueException ex)
                    {
                        OperationalLogs.LogMessageQueueFatalError(Helpers.GetSetting<string>("NotificationQueueServer",null), ex);
                        throw;
                    }
                }
            }
        }

        /// <summary>
        /// Called asynchronously when a notification item has been received from the message queue.  This is
        /// where processing of a notification from the message queue begins.
        /// </summary>
        private void OnMessageReceived(IAsyncResult asyncResult)
        {
            _outstandingRead = false;

            // Ensure that the service isn't shutdown until this operation completes
            if (!IncrementRefCount())
            {
                // Don't receive the message from the response queue because the service is stopping
                return;
            }

            try
            {
                Message message = _queue.EndReceive(asyncResult);
                Debug.Assert(message != null, "message != null");
                NotificationItem notification = ParseNotificationItem(message);

                // once we've finished receiving this message, go start listening again immediately
                Logger.WriteTraceInformation(notification.Context, "Message received, listening for new message.");
                bool canRead = false;

                lock (_stateSyncRoot)
                {
                    if (!_outstandingRead)
                    {
                        _outstandingRead = true;
                        canRead = true;
                    }
                }

                if (canRead)
                {
                    _queue.BeginReceive(TimeSpan.FromMinutes(1.0), this, new AsyncCallback(OnMessageReceived));
                }

                if (!_responseLocations.ContainsKey(notification.Id))
                {
                    LogResponseQueue(message, notification);
                    DeliverNotification(notification);
                }
                else
                {
                    Logger.WriteTraceWarning("Attempted to deliver two notifications with the same ID. Ignoring the second one.");
                    DecrementRefCount();
                }
            }
            catch (MessageQueueException ex)
            {
                DecrementRefCount();

                // Ignore errors due to an empty queue as these are expected if this callback is invoked because of a timeout.
                if (ex.MessageQueueErrorCode != MessageQueueErrorCode.MessageNotFound && ex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
                {
                    OperationalLogs.LogMessageQueueFatalError(Helpers.GetSetting<string>("NotificationQueueServer", null), ex);

                    // stop the service; we can't do anything if we can't read from MSMQ.
                    Stop();
                }
                else
                {
                    _queue.BeginReceive(TimeSpan.FromMinutes(1.0), this, new AsyncCallback(OnMessageReceived));
                }
            }
        }

        /// <summary>
        /// Unregisters any SipEndpoints in use and stops listening for further
        /// messages on MSMQ. The service will keep running until all currently
        /// running delivery classes are complete.
        /// </summary>
        protected override void OnStop()
        {
            OperationalLogs.LogServiceStoppingInformation();

            Debug.Assert(!_stopping && _waitEvent == null, "OnStop should not be called when service is already shutting down.");

            lock (_stateSyncRoot)
            {
                _stopping = true;

                if (_refCount > 0)
                {
                    _waitEvent = new AutoResetEvent(false);
                }
            }

            if (_waitEvent != null)
            {
                if (!_waitEvent.WaitOne(Helpers.GetSetting<int>("ShutdownTimeoutInSeconds",30) * 1000, false))
                {
                    OperationalLogs.LogServiceShutdownTimeoutWarning(Helpers.GetSetting<int>("ShutdownTimeoutInSeconds",30));
                    _waitEvent.Close();
                }
            }

            _factory.Uninitialize();
            _factory.Dispose();
            _queue.Dispose();

            OperationalLogs.LogServiceStoppedInformation();
        }

        /// <summary>
        /// Update the list of response queues in use.
        /// </summary>
        private void LogResponseQueue(Message message, NotificationItem notification)
        {
            Debug.Assert(notification != null, "notification != null");
            Debug.Assert(notification.Id != null, "notificiation.Id != null");

            lock (_stateSyncRoot)
            {
                Debug.Assert(!_responseLocations.ContainsKey(notification.Id), "Received two notifications with same ID.");

                if (message.ResponseQueue != null)
                {
                    _responseLocations.Add(notification.Id, message.ResponseQueue.Path);
                }
                else
                {
                    _responseLocations.Add(notification.Id, null);
                }
            }
        }

        /// <summary>
        /// Deliver the notification.
        /// </summary>
        private void DeliverNotification(NotificationItem notification)
        {
            _dispatcher.DeliverNotificationAsync(notification);
        }

        /// <summary>
        /// Handler for notification delivery completed event.
        /// </summary>
        private void OnNotificationCompleted(object sender, DeliveryCompletedEventArgs e)
        {
            Debug.Assert(e != null, "Delivery completed arguments cannot be null.");
            Debug.Assert(e.Result != null, "Delivery completed event must contain a result.");
            Debug.Assert(e.Result.Id != null, "e.Result.Id != null");
            Debug.Assert(_refCount > 0, "OnNotificationCompleted called with a reference count of 0");

            WriteResponseToQueue(e);

            DecrementRefCount();

            lock (_stateSyncRoot)
            {
                _responseLocations.Remove(e.Result.Id);

                if (_stopping && _refCount <= 0)
                {
                    Helpers.AssertCast<NotificationDispatcher>(sender).DeliverNotificationCompleted -= OnNotificationCompleted;
                }
            }
        }

        /// <summary>
        /// Writes a message to the response queue, if one is available, notifying interested 
        /// parties that delivery is complete.
        /// </summary>
        private void WriteResponseToQueue(DeliveryCompletedEventArgs e)
        {
            Debug.Assert(e != null, "e != null");
            Debug.Assert(e.Result != null, "e.Result != null");
            Debug.Assert(e.Result.Context != null, "e.Result.Context != null"); 
            Debug.Assert(e.Result.Id != null, "e.Result.Id != null");

            string responseQueuePath = null;

            lock (_stateSyncRoot)
            {
                Debug.Assert(_responseLocations.ContainsKey(e.Result.Id), "Received response for unknown notification ID");

                responseQueuePath = _responseLocations[e.Result.Id];
            }

            if (responseQueuePath != null)
            {
                MessageQueue responseQueue = new MessageQueue(responseQueuePath);

                if (responseQueue.CanWrite)
                {
                    Logger.WriteTraceInformation(e.Result.Context, "Writing MSMQ response for notification ID: " + e.Result.Id.ToString());
                    Message response = new Message(e.Result, new BinaryMessageFormatter());

                    responseQueue.Send(response);
                }
                else
                {
                    OperationalLogs.LogMessageQueueError(responseQueue.Path, null);
                }
            }
            else
            {
                Logger.WriteTraceWarning(e.Result.Context, "There is no response queue to write to for notification ID: " + e.Result.Id.ToString());
            }
        }

        /// <summary>
        /// Parses the NotificationItem out of the message received from MSMQ.
        /// </summary>
        private static NotificationItem ParseNotificationItem(Message message)
        {
            message.Formatter = new BinaryMessageFormatter();
            Debug.Assert(message.Body != null, "Message body should not be null.");

            return Helpers.AssertCast<NotificationItem>(message.Body);
        }

        /// <summary>
        /// Increment the count of current outstanding notifications
        /// </summary>
        /// <remarks>
        /// The reference counting ensures that the service does not shutdown part way through an MSMQ-related operation.
        /// </remarks>
        /// <returns>
        /// true if the service is not stopping; false otherwise.
        /// </returns>
        private bool IncrementRefCount()
        {
            lock (_stateSyncRoot)
            {
                if (_stopping || !_connected)
                {
                    return false;
                }
                else
                {
                    _refCount++;
                }
            }
            return true;
        }

        /// <summary>
        /// Decrement the count of the current outstanding notifications
        /// </summary>
        /// <remarks>
        /// The reference counting ensures that the service does not shutdown part way through an MSMQ-related operation.
        /// </remarks>
        private void DecrementRefCount()
        {
            lock (_stateSyncRoot)
            {
                _refCount--;

                if (_stopping && _refCount <= 0)
                {
                    Debug.Assert(_waitEvent != null, "_waitEvent != null");

                    _waitEvent.Set();
                }
            }
        }
    }
}
